#include <fcntl.h>
#include <unistd.h>

#include <mutex>
#include <sstream>

#include "debug.hpp"
#include "misc.hpp"
#include "sharedMem.hpp"


// 发送文件描述符辅助函数
bool sharedMem::send_fd__(int socket, int fd) {
    struct msghdr msg = {0};
    char buf[CMSG_SPACE(sizeof(fd))];
    memset(buf, 0, sizeof(buf));
    
    // 在辅助数据中发送文件描述符
    struct iovec io = { .iov_base = (void*)"", .iov_len = 1 };
    
    msg.msg_iov = &io;
    msg.msg_iovlen = 1;
    msg.msg_control = buf;
    msg.msg_controllen = sizeof(buf);
    
    struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
    cmsg->cmsg_level = SOL_SOCKET;
    cmsg->cmsg_type = SCM_RIGHTS;
    cmsg->cmsg_len = CMSG_LEN(sizeof(fd));
    
    *(int *)CMSG_DATA(cmsg) = fd;
    
    msg.msg_controllen = cmsg->cmsg_len;
    
    if (sendmsg(socket, &msg, 0) < 0) {
        ERROR_MSG( "发送数据失败" );
        return false;
    }
    
    return true;
}

// 接收文件描述符辅助函数
int sharedMem::recv_fd__(int socket) {
    struct msghdr msg = {0};
    
    char m_buffer[256];
    struct iovec io = { .iov_base = m_buffer, .iov_len = sizeof(m_buffer) };
    msg.msg_iov = &io;
    msg.msg_iovlen = 1;
    
    char c_buffer[256];
    msg.msg_control = c_buffer;
    msg.msg_controllen = sizeof(c_buffer);
    
    if (recvmsg(socket, &msg, 0) < 0) {
        ERROR_MSG("读取数据失败");
        return -1;
    }
    
    struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
    if (cmsg == NULL) {
        ERROR_MSG("没有收到数据");
        return -1;
    }
    
    if (cmsg->cmsg_level != SOL_SOCKET || cmsg->cmsg_type != SCM_RIGHTS) {
        ERROR_MSG("接收到的辅助数据不是文件描述符类型（SCM_RIGHTS）");
        return -1;
    }
    
    int fd;
    memcpy(&fd, CMSG_DATA(cmsg), sizeof(fd));
    return fd;
}

// 初始化 epoll
bool sharedMem::init_epoll() {
    m_epoll_fd__ = epoll_create1(0);
    if (m_epoll_fd__ == -1) {
        ERROR_MSG("创建epoll实例失败");
        return false;
    }
    
    // 添加eventfd到epoll监听
    struct epoll_event event;
    event.events = EPOLLIN | EPOLLET; // 边缘触发模式
    event.data.fd = m_eventfd__;
    
    if (epoll_ctl(m_epoll_fd__, EPOLL_CTL_ADD, m_eventfd__, &event) == -1) {
        ERROR_MSG("添加eventfd到epoll失败");
        close(m_epoll_fd__);
        m_epoll_fd__ = -1;
        return false;
    }
    
    return true;
}

// 事件循环
void sharedMem::event_loop__() {
    const int MAX_EVENTS = 10;
    struct epoll_event events[MAX_EVENTS];
    
    while (m_running__) {
        int nfds = epoll_wait(m_epoll_fd__, events, MAX_EVENTS, 5); // 5ms超时
        if (nfds == -1) {
            if (errno == EINTR) {
                continue; // 被信号中断，继续等待
            }
            ERROR_MSG("epoll_wait失败");
            break;
        }
        
        for (int i = 0; i < nfds; i++) {
            if (events[i].data.fd == m_eventfd__) {
                // eventfd可读事件
                uint64_t value;
                ssize_t s = read(m_eventfd__, &value, sizeof(value));
                if (s == sizeof(value) && m_event_callback__) {
                    // 调用事件回调
                    m_event_callback__(value);
                }
            }
        }
    }
}


semaphore :: semaphore( const std::string& name , bool svr ):
        m_name__( name ), p_sem__( nullptr )
{
        if( svr ){
                p_sem__ = sem_open( name.c_str() , O_CREAT | O_EXCL, 0666 , 1 );
        }else{
                p_sem__ = sem_open( name.c_str() , 0 );
        }
        if( !p_sem__ ){
                const char * msg = strerror( errno );
                if( msg ){
                        std::string str( msg );
                        throw std::runtime_error( msg );
                }else{
                        throw std::runtime_error( "创建信号量失败" );
                }
        }
}

semaphore :: ~semaphore()
{
        MSG( "关闭共享内存" , TNORMAL );
        std::lock_guard< std::mutex >   lck( m_mutex__ );
        if( p_sem__ ){
                sem_close( p_sem__ );
                sem_unlink( m_name__.c_str() );
        }
}

bool semaphore :: lock( bool sw )
{
        std::lock_guard< std::mutex >  lck( m_mutex__ );
        int rc = 0;
        if( sw ){
                rc = sem_wait( p_sem__ );
        }else{
                rc = sem_post( p_sem__ );
        }

        if( rc ){
                ERROR_MSG( strerror( errno ) );
                return false;
        }

        return true;
}

bool semaphore :: tryLock( bool sw )
{
        std::lock_guard< std::mutex >  lck( m_mutex__ );
        int rc = 0;
        if( sw ){
                rc = sem_trywait( p_sem__ );
        }else{
                rc = sem_post( p_sem__ );
        }

        if( rc ){
                ERROR_MSG( strerror( errno ) );
                return false;
        }

        return true;
}
////////////////////////////////////////////////////////////////////////////////////////////////////
smfLockGuard :: smfLockGuard( semaphore& sm ): m_sem__( sm )
{
        sm.lock( true );
}

smfLockGuard :: ~smfLockGuard()
{
        m_sem__.unlock();
}

////////////////////////////////////////////////////////////////////////////////////////////////////
sharedMem :: sharedMem( const std::string& name , uint64_t size , bool svr ):
        m_name__( name ) ,
        m_size__( size ) ,
        m_data_len__( 0 ),
        m_smf__ ( name + "_smf" , svr ),
        m_fd__( -1 ),
		m_eventfd__( -1 ),
        m_socket_path__( socket_path.empty() ? "/tmp/" + name + "_socket" : socket_path ),
        m_socket_fd__( -1 ),
        m_is_server__( svr ),
        m_epoll_fd__( -1 ),
        m_running__( false ),
        m_event_thread__()
{
        if( svr ){
                m_fd__ = shm_open( name.c_str() , O_CREAT | O_RDWR , 0666 );
        }else{
                m_fd__ = shm_open( name.c_str() , O_RDWR , 0666 );
        }
        
        if( m_fd__ < 0 ){
                throw std::runtime_error( strerror( errno ));
        }

        ftruncate( m_fd__ , size );

        p_shared_mem__ = ( uint8_t * )mmap( NULL , size , PROT_READ | PROT_WRITE , MAP_SHARED , m_fd__ , 0 );
        if( !p_shared_mem__ ){
                throw std::runtime_error( "将共享内存映射到本地内存失败" );
        }
		// 处理eventfd和Unix域套接字
        if (svr) {
            // 服务器端：创建eventfd和Unix域套接字服务器
            m_eventfd__ = eventfd(0, EFD_NONBLOCK);
            if (m_eventfd__ == -1) {
                throw std::runtime_error("创建eventfd失败");
            }
            
            // 创建Unix域套接字服务器
            m_socket_fd__ = socket(AF_UNIX, SOCK_STREAM, 0);
            if (m_socket_fd__ == -1) {
                throw std::runtime_error("创建Unix域套接字失败");
            }
            
            // 绑定套接字
            struct sockaddr_un addr;
            memset(&addr, 0, sizeof(addr));
            addr.sun_family = AF_UNIX;
            strncpy(addr.sun_path, m_socket_path__.c_str(), sizeof(addr.sun_path) - 1);
            
            // 确保文件不存在
            unlink(m_socket_path__.c_str());
            
            if (bind(m_socket_fd__, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
                throw std::runtime_error("绑定Unix域套接字失败");
            }
            
            if (listen(m_socket_fd__, 5) == -1) {
                throw std::runtime_error("监听Unix域套接字失败");
            }
            
            // 在新线程中接受连接并发送eventfd
            std::thread([this]() {
                int client_fd = accept(m_socket_fd__, NULL, NULL);
                if (client_fd == -1) {
                    ERROR_MSG("接受客户端连接失败");
                    return;
                }
                
                if (!send_fd__(client_fd, m_eventfd__)) {
                    ERROR_MSG("发送eventfd失败");
                }
                
                close(client_fd);
            }).detach();
		} else {
            // 客户端：连接Unix域套接字服务器并接收eventfd
            m_socket_fd__ = socket(AF_UNIX, SOCK_STREAM, 0);
            if (m_socket_fd__ == -1) {
                throw std::runtime_error("创建Unix域套接字失败");
            }
            
            struct sockaddr_un addr;
            memset(&addr, 0, sizeof(addr));
            addr.sun_family = AF_UNIX;
            strncpy(addr.sun_path, m_socket_path__.c_str(), sizeof(addr.sun_path) - 1);
            
            // 尝试连接，可能需要重试几次
            int retries = 5;
            while (connect(m_socket_fd__, (struct sockaddr*)&addr, sizeof(addr)) == -1 && retries > 0) {
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                retries--;
            }
            
            if (retries == 0) {
                throw std::runtime_error("连接Unix域套接字服务器失败");
            }
            
            // 接收eventfd
            m_eventfd__ = recv_fd__(m_socket_fd__);
            if (m_eventfd__ == -1) {
                throw std::runtime_error("接收eventfd失败");
            }
            
            close(m_socket_fd__);
            m_socket_fd__ = -1;
            
            // 客户端初始化epoll
            if (!init_epoll()) {
                throw std::runtime_error("初始化epoll失败");
            }
        }
}

sharedMem :: ~sharedMem()
{
        smfLockGuard lck( m_smf__ );

        close( m_fd__ );
        shm_unlink( m_name__.c_str() );
		
		// 关闭eventfd
        if (m_eventfd__ != -1) {
            close(m_eventfd__);
        }
        
        // 关闭Unix域套接字并清理
        if (m_socket_fd__ != -1) {
            close(m_socket_fd__);
        }
        
        // 关闭epoll
        if (m_epoll_fd__ != -1) {
            close(m_epoll_fd__);
        }
        
        if (m_is_server__) {
            unlink(m_socket_path__.c_str());
        }
}

size_t sharedMem :: send( const uint8_t * data ,size_t len )
{
        smfLockGuard lck( m_smf__ );
        std::stringstream ss;
        if( len < m_size__ ){
                m_data_len__ = len;
                ss << m_data_len__ << ",";
                std::string str = ss.str();
                memcpy( p_shared_mem__ , str.c_str() , str.length() );
                memcpy( p_shared_mem__ + str.length() , data , len );
        }else{
                m_data_len__ = len;
                ss << m_data_len__ << ",";
                std::string str = ss.str();
                m_data_len__ = m_size__ - str.length();
                memcpy( p_shared_mem__ + str.length(), data , m_data_len__ );
                
        }

        return m_data_len__;
}

size_t sharedMem :: recv( uint8_t * data , size_t len )
{
        smfLockGuard lck( m_smf__ );
        char * s = ( char *)p_shared_mem__;
        char * e = strstr( s , "," );
        std::string str( s , e - s );

        std::stringstream ss;
        ss << str;
        ss >> m_data_len__;

        if( len > m_data_len__ ){
                memcpy( data , p_shared_mem__ + str.length() + 1, m_data_len__ );
                return m_data_len__;
        }
        memcpy( data , p_shared_mem__ + str.length() + 1 , len );
                
        return len;
}


bool sharedMem::notify(uint64_t value) {
    if (m_eventfd__ == -1) {
        return false;
    }
    
    ssize_t s = write(m_eventfd__, &value, sizeof(value));
    return s == sizeof(value);
}

uint64_t sharedMem::wait() {
    if (m_eventfd__ == -1) {
        return 0;
    }
    
    uint64_t value;
    ssize_t s = read(m_eventfd__, &value, sizeof(value));
    if (s != sizeof(value)) {
        return 0;
    }
    
    return value;
}

void sharedMem::setEventCallback(std::function<void(uint64_t)> callback) {
    m_event_callback__ = callback;
}

bool sharedMem::startEventLoop() {
    if (m_is_server__ || m_epoll_fd__ == -1) {
        return false; // 服务器端不需要事件循环，或者epoll未初始化
    }
    
    if (m_running__) {
        return true; // 事件循环已经在运行
    }
    
    m_running__ = true;
    m_event_thread__ = std::thread(&sharedMem::event_loop__, this);
    
    return true;
}

void sharedMem::stopEventLoop() {
    if (!m_running__) {
        return;
    }
    
    m_running__ = false;
    if (m_event_thread__.joinable()) {
        m_event_thread__.join();
    }
}